package com.deep.cdc;

import com.deep.bean.ConnectConfig;
import com.deep.bean.TransformConfig;
import com.deep.common.ArgsConfig;
import com.deep.common.ArgsUtil;
import com.deep.common.ProfileConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * t_wf_accreditinst
 */
@Slf4j
public class SyncAccreditInst {
    public static void main(String[] args) {
        ArgsConfig argsConfig = ArgsUtil.getMap(args);
        Integer recid = argsConfig.getRecid();
        TransformConfig transformConfig = ProfileConfig.getTransformConfig(argsConfig.getProfile());
        ConnectConfig sourceConfig = transformConfig.getSource();
        ConnectConfig targetConfig = transformConfig.getTarget();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        //创建用于输出的表
        String createInputDDL = "create table source_t_wf_accreditinst  " +
                "(   " +
                "RECID  int  ," +
                "DETAILNO  int  ," +
                "PROCID  int  ," +
                "WORKITEMID  int  ," +
                "APPLYACTID  int  ," +
                "APPLYITEMID  int  ," +
                "APPLYPARTID  int  ," +
                "APPLYPARTNAME  string  ," +
                "REPLYACTID  int  ," +
                "REPLYITEMID  int  ," +
                "REPLYPARTID  int  ," +
                "REPLYPARTNAME  string  ," +
                "ACCREDITSTATEID  int  ," +
                "APPLYTYPEID  int  ," +
                "APPLYSHOWINFO  string  ," +
                "APPLYMEMO  string  ," +
                "APPLYDATE  timestamp(0)  ," +
                "TIMEUNIT  int  ," +
                "APPLYTIME  double  ," +
                "REPLYTYPEID  int  ," +
                "REPLYMEMOFLAG  int  ," +
                "REPLYMEMO  string  ," +
                "REPLYDATE  timestamp(0)  ," +
                "REPLYROLEPARTID  int  ," +
                "REPLYROLEPARTNAME  string  ," +
                "    PRIMARY KEY (`DETAILNO`) NOT ENFORCED  " +
                ")WITH (  " +
                "    'connector' = 'mysql-cdc',  " +
                "    'hostname' = '" + sourceConfig.getHostname() + "',  " +
                "    'port' = '" + sourceConfig.getPort() + "',   " +
                "    'username' = '" + sourceConfig.getUsername() + "',  " +
                "    'password' = '" + sourceConfig.getPassword() + "',   " +
                "    'database-name' = '" + sourceConfig.getDatabaseName() + "',  " +
                "    'table-name' = 't_wf_accreditinst'," +
                "    'server-time-zone' = 'Asia/Shanghai', " +
                "   'scan.startup.mode' = 'initial', " +
                "   'debezium.skipped.operations'='d'  " +
                ") ";
        log.info("createInputDDL = {}", createInputDDL);

        tableEnv.executeSql(createInputDDL);


        //创建用于输出的表
        String createOutputDDL = "create table target_t_wf_accreditinst " +
                "( " +
                "RECID  int  ," +
                "DETAILNO  int  ," +
                "PROCID  int  ," +
                "WORKITEMID  int  ," +
                "APPLYACTID  int  ," +
                "APPLYITEMID  int  ," +
                "APPLYPARTID  int  ," +
                "APPLYPARTNAME  string  ," +
                "REPLYACTID  int  ," +
                "REPLYITEMID  int  ," +
                "REPLYPARTID  int  ," +
                "REPLYPARTNAME  string  ," +
                "ACCREDITSTATEID  int  ," +
                "APPLYTYPEID  int  ," +
                "APPLYSHOWINFO  string  ," +
                "APPLYMEMO  string  ," +
                "APPLYDATE  timestamp(0)  ," +
                "TIMEUNIT  int  ," +
                "APPLYTIME  double  ," +
                "REPLYTYPEID  int  ," +
                "REPLYMEMOFLAG  int  ," +
                "REPLYMEMO  string  ," +
                "REPLYDATE  timestamp(0)  ," +
                "REPLYROLEPARTID  int  ," +
                "REPLYROLEPARTNAME  string  ," +
                "    PRIMARY KEY (`DETAILNO`) NOT ENFORCED   " +
                ")WITH ( " +
                " 'connector' = 'jdbc',  " +
                " 'driver' = 'com.mysql.cj.jdbc.Driver', " +
                " 'url' = 'jdbc:mysql://"+ targetConfig.getHostname() + ":" + targetConfig.getPort() + "/" + targetConfig.getDatabaseName() + "?rewriteBatchedStatements=true&useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai',  " +
                "    'username' = '" + targetConfig.getUsername() + "',  " +
                "    'password' = '" + targetConfig.getPassword() + "',   " +
                "  'table-name' = 't_wf_accreditinst',  " +
                "  'connection.max-retry-timeout' = '60s'  " +
                ")";
        log.info("createOutputDDL = {}", createOutputDDL);

        tableEnv.executeSql(createOutputDDL);


        TableResult tableResult = tableEnv.executeSql("insert into target_t_wf_accreditinst select * from source_t_wf_accreditinst where RECID >  " + recid);

        tableResult.print();

    }
}
